package com.atguigu.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseApp;
import com.atguigu.realtime.common.ConstantTopic;
import com.atguigu.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;

/**
 * @ClassName: DWDLogApp
 * @Description:
 * @Author: kele
 * @Date: 2021/4/17 10:15
 **/
public class DWDLogApp extends BaseApp {

    public static void main(String[] args) {

        new DWDLogApp().init(20000,2,"DWDLogApp","ods_log","DWDLogApp");

    }

    /**
     * @param env
     * @param dataStreamSource
     *
     *  1、确认新老用户，将新用户的is_new", "1"，老用户的is_new", "0"
     *  2、分流，将 启动日志， 曝光日志， 页面日志分成3个流
     */
    @Override
    protected void run(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {

        // 区分新老用户
        SingleOutputStreamOperator<JSONObject> jsonData = distinguishNewOrOld(dataStreamSource);

        /**
         * 分流
         * tuple0 ：启动日志
         * tuple1 ：页面日志
         * tuple2 : 曝光日志
         */
        Tuple3<DataStream<JSONObject>, DataStream<JSONObject>, DataStream<JSONObject>> streamTuple3 = splitStream(jsonData);

        //将数据写入到kafka中
        sendToKafka(streamTuple3);
    }


    /**
     * 区分新老用户
     *
     * 1、加水印，将用户按照mid分组
     * 2、开窗，
     * 3、将第一个窗中的数据按时间排序，第一个设置为新用户
     * 4、将之后的数据中is_new" 设置为 "0"
     */
    private SingleOutputStreamOperator<JSONObject> distinguishNewOrOld(DataStreamSource<String> dataStream) {

        return dataStream
                //将字符串转为jsonobject
                .map(JSONObject::parseObject)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.
                                <JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return element.getLong("ts");
                            }
                        })
                )
                .keyBy(value -> value.getJSONObject("common").getString("mid"))
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                /**
                 *
                 */
                .process(new ProcessWindowFunction<JSONObject, JSONObject, String, TimeWindow>() {

                    private ValueState<Long> firstState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        firstState = getRuntimeContext()
                                .getState(new ValueStateDescriptor<>("firstState", Long.class));
                    }

                    @Override
                    public void process(String s,
                                        Context context,
                                        Iterable<JSONObject> elements,
                                        Collector<JSONObject> out) throws Exception {

                        /**
                         * 第一个窗口的数据到达
                         * 1、数据放到list集合中排序
                         */
                        //是将状态的值取出判断是否为空
                        if(firstState.value() == null){

                            ArrayList<JSONObject> list = new ArrayList<>();
                            for (JSONObject element : elements) {
                                list.add(element);
                            }
                            //对集合中的数据进行排序
                            list.sort(Comparator.comparing(o -> o.getLong("ts")));

                            //将第一个元素设置为新用户，其他的遍历输出,更新状态
                            for (int i = 0; i < list.size(); i++) {
                                if(i == 0){
                                    list.get(i).getJSONObject("common").put("is_new", "1");
                                    firstState.update(list.get(0).getLong("ts"));
                                }

                                else{
                                    list.get(i).getJSONObject("common").put("is_new", "0");
                                }

                                    out.collect(list.get(i));
                            }
                            //之后的数据到达,将类型都设置为旧用户
                        }else{
                            for (JSONObject element : elements) {
                                element.getJSONObject("common").put("is_new","0");
                                out.collect(element);
                            }
                        }
                    }
                });
    }

    /**
     * 分流
     * 主流：启动日志
     * 侧输出流：曝光日志，页面日志
     *      -- 页面日志中有page，曝光日志中有dispaly字段，同时也有page字段
     * 将三个流通过元组的方式返回
     *
     * @param stream
     */
    private Tuple3<DataStream<JSONObject>, DataStream<JSONObject>, DataStream<JSONObject>>
                        splitStream(SingleOutputStreamOperator<JSONObject> stream) {

        OutputTag<JSONObject> pageInfo = new OutputTag<JSONObject>("pageInfo") {};
        OutputTag<JSONObject> displayInfo = new OutputTag<JSONObject>("displays") {};

        SingleOutputStreamOperator<JSONObject> main = stream.process(new ProcessFunction<JSONObject, JSONObject>() {

            /**
             * 将启动日志放入主流中
             * 曝光日志，页面日志放到侧输出流中
             * @param value
             * @param ctx
             * @param out
             * @throws Exception
             */
            @Override
            public void processElement(JSONObject value,
                                       Context ctx,
                                       Collector<JSONObject> out) throws Exception {

                //将数据分类，获取三种数据
                JSONObject startlog = value.getJSONObject("start");
                JSONObject pagelog = value.getJSONObject("page");
                JSONArray displaylog = value.getJSONArray("displays");

                if (startlog != null) {
                    out.collect(value);
                } else {

                    if (pagelog != null)
                        ctx.output(pageInfo, value);

                    /**
                     * 如果是曝光日志，是数组的形式，只需要其中的每个信息（曝光的产品对我们有用）
                     */
                    if (displaylog != null && displaylog.size() > 0) {
                        for (int i = 0; i < displaylog.size(); i++) {
                            JSONObject data = displaylog.getJSONObject(i);

                            data.put("mid", value.getJSONObject("common").getString("mid"));
                            data.put("ts", value.getString("ts"));
                            data.put("page_id", value.getJSONObject("page").getString("page_id"));

                            ctx.output(displayInfo, data);
                        }
                    }
                }

            }
        });

        DataStream<JSONObject> page = main.getSideOutput(pageInfo);
        DataStream<JSONObject> display = main.getSideOutput(displayInfo);
        return Tuple3.of(main,page,display);

    }

    private void sendToKafka(Tuple3<DataStream<JSONObject>, DataStream<JSONObject>, DataStream<JSONObject>> streams) {

        streams
                .f0
                .map(obj -> JSON.toJSONString(obj))
                .addSink(MyKafkaUtil.getKafkaSink(ConstantTopic.DWD_START));

        streams
                .f1
                .map(obj -> JSON.toJSONString(obj))
                .addSink(MyKafkaUtil.getKafkaSink(ConstantTopic.DWD_PAGE));

        streams
                .f2
                .map(obj -> JSON.toJSONString(obj))
                .addSink(MyKafkaUtil.getKafkaSink(ConstantTopic.DWD_DISPLAY));
    }

}
